package publisherconfirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: LianBao
 * Date: 2025-03-27
 * Time: 20:08
 */
public class PublisherConfirms {
    private static final Integer MESSAGE_COUNT = 10000;

    public static Connection createConnection() throws IOException, TimeoutException {
        //1. 建立连接(Connection)
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置连接参数:ip,端口号,用户名,密码,虚拟机
        connectionFactory.setHost("116.205.165.92");//ip
        connectionFactory.setPort(5672);//端口号,默认是5672
        connectionFactory.setUsername("Test");//用户名
        connectionFactory.setPassword("123");//密码
        connectionFactory.setVirtualHost("TestVirtualHost");//设置操作的虚拟机
        //从连接工厂中获取到连接
        return connectionFactory.newConnection();
    }

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        //1.单独确认
//        publishMessagesIndividually();
        //2.批量确认
        publishMessagesInBatches();
        //3.异步确认
        handlingPublisherConfirmsAsynchronously();

    }

    //1.单独确认,每发送一条消息,就等待确认,确认完毕再发送下一条消息
    private static void publishMessagesIndividually() throws IOException, TimeoutException, InterruptedException {
        try (Connection connection = createConnection()) {
            //1.开启信道
            Channel channel = connection.createChannel();
            //2.设置信道为confirm模式,开启之后,基于该信道发送的消息都会生成一个id,用于消息的确认
            channel.confirmSelect();
            //3.声明队列
            channel.queueDeclare("PUBLISHER_CONFIRM_QUEUE1", true, false, false, null);
            //4.发送消息并等待确认
            long start = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "HelloMessage" + i;
                channel.basicPublish("", "PUBLISHER_CONFIRM_QUEUE1", null, msg.getBytes());
                //等待确认
                channel.waitForConfirmsOrDie(5000);//如果超时了会抛出异常
            }
            long end = System.currentTimeMillis();
            System.out.println("批量确认策略,消息条数" + MESSAGE_COUNT + "耗时" + (end - start) + "ms");

        }
    }


    //2.批量确认
    private static void publishMessagesInBatches() throws IOException, TimeoutException, InterruptedException {
        try (Connection connection = createConnection()) {
            //1.开启信道
            Channel channel = connection.createChannel();
            //2.设置信道为confirm模式,开启之后,基于该信道发送的消息都会生成一个id,用于消息的确认
            channel.confirmSelect();
            //3.声明队列
            channel.queueDeclare("PUBLISHER_CONFIRM_QUEUE2", true, false, false, null);
            //4.发送消息并等待确认
            int count = 0;
            int batchSize = 50;//批量处理的消息数量
            long start = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "HelloMessage" + i;
                channel.basicPublish("", "PUBLISHER_CONFIRM_QUEUE2", null, msg.getBytes());
                count++;
//                等待确认
                if (count == batchSize) {//如果发送的消息达到50条,就批量确认
                    channel.waitForConfirmsOrDie(5000);//如果超时了会抛出异常
                    count = 0;
                }
            }
            //如果count>0再整体进行确认
            if (count > 0) {
                channel.waitForConfirmsOrDie(5000);//如果超时了会抛出异常
            }
            long end = System.currentTimeMillis();
            System.out.println("批量确认策略,消息条数" + MESSAGE_COUNT + "耗时" + (end - start) + "ms");

        }

    }

    //3.异步确认
    private static void handlingPublisherConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException {
        try (Connection connection = createConnection()) {
            //1.开启信道
            Channel channel = connection.createChannel();
            //2.设置信道为confirm模式,开启之后,基于该信道发送的消息都会生成一个id,用于消息的确认
            channel.confirmSelect();
            //3.声明队列
            channel.queueDeclare("PUBLISHER_CONFIRM_QUEUE3", true, false, false, null);

            //
            long start = System.currentTimeMillis();
            SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());//存储未确认的消息的id
            //4.异步监听
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    //处理确认
                    if (multiple) {
                        //如果是批量处理,清除掉当前id之前的消息
                        confirmSeqNo.headSet(deliveryTag + 1).clear();
                    } else {
                        //如果不是批量处理,清除当前的这个id就可以了
                        confirmSeqNo.remove(deliveryTag);
                    }
                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    //处理
                    if (multiple) {
                        //如果是批量处理,清除掉当前id之前的消息
                        confirmSeqNo.headSet(deliveryTag + 1).clear();
                    } else {
                        //如果不是批量处理,清除当前的这个id就可以了
                        confirmSeqNo.remove(deliveryTag);
                    }
                    //重发消息
                }
            });
            //5.发送消息
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String msg = "HelloMessage" + i;
                long id = channel.getNextPublishSeqNo();
                channel.basicPublish("", "PUBLISHER_CONFIRM_QUEUE3", null, msg.getBytes());
                confirmSeqNo.add(id);//id加入集合中,表示消息发送完了,但是还没确认
            }
            while (!confirmSeqNo.isEmpty()) {//如果存储id的集合消息不为空,说明还有消息没有确认
                Thread.sleep(10);
            }
            long end = System.currentTimeMillis();
            System.out.println("异步确认策略,消息条数" + MESSAGE_COUNT + "耗时" + (end - start) + "ms");


        }
    }

}
